package com.alison.sparkstream.action

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import java.util.Properties
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
import scala.util.Random

object E1_generate_data {
  """
    |每条记录的内容格式如下：
    |时间戳 地区名称 城市名称 用户id 广告id
    |1692273126289 华东 上海 4 2
    |1692273126289 华北 天津 2 1
    |1692273126289 华北 天津 2 3
    |1692273126289 华南 深圳 2 5
    |1692273126289 华南 深圳 1 5
    |
    |""".stripMargin

  """
    |启动后可以看到程序每隔 2 秒便会产生一批数据
    |""".stripMargin
  // 创建 Kafka 生产者
  def createKafkaProducer(broker: String, topic: String): KafkaProducer[String, String] = {
    // 创建配置对象
    val props = new Properties()
    props.put("bootstrap.servers", broker)
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    new KafkaProducer[String, String](props)
  }

  def main(args: Array[String]): Unit = {
    val topic = "spark-topic" // Kafka主题
    val broker = "localhost:9092" // Kafka集群地址
    // 创建 Kafka 消费者
    val kafkaProducer: KafkaProducer[String, String] = createKafkaProducer(broker, topic)
    while (true) {
      // 随机产生实时数据并通过 Kafka 生产者发送到 Kafka 集群中
      for (line <- generateMockData()) {
        kafkaProducer.send(new ProducerRecord[String, String](topic, line))
        println(line)
      }
      Thread.sleep(2000) // 休眠2秒
    }
  }
  // 生成模拟实时数据
  def generateMockData(): Array[String] = {
    val array: ArrayBuffer[String] = ArrayBuffer[String]()

    // 城市信息随机选项集合
    val CityRandomOpt = RandomOptions(
      RanOpt(CityInfo(1, "北京", "华北"), 30),
      RanOpt(CityInfo(2, "上海", "华东"), 30),
      RanOpt(CityInfo(3, "广州", "华南"), 10),
      RanOpt(CityInfo(4, "深圳", "华南"), 20),
      RanOpt(CityInfo(5, "天津", "华北"), 10)
    )
    val random = new Random()
    // 模拟实时数据：
    // timestamp province city userid adid
    for (i <- 0 to 50) {
      val timestamp: Long = System.currentTimeMillis()
      val cityInfo: CityInfo = CityRandomOpt.getRandomOpt
      val city: String = cityInfo.city_name
      val area: String = cityInfo.area
      val adid: Int = 1 + random.nextInt(6)
      val userid: Int = 1 + random.nextInt(6)
      // 拼接实时数据
      array += timestamp + " " + area + " " + city + " " + userid + " " + adid
    }
    array.toArray
  }
  /**
   * 城市信息
   *
   * @param city_id   城市 id
   * @param city_name 城市名称
   * @param area      城市所在大区
   */
  case class CityInfo(city_id: Long, city_name: String, area: String)

  // 定义一个带权重的随机选项数据结构
  case class RanOpt[T](value: T, weight: Int)

  // 伴生对象，用于创建随机选项集合
  object RandomOptions {
    def apply[T](opts: RanOpt[T]*): RandomOptions[T] = {
      val randomOptions = new RandomOptions[T]()
      for (opt <- opts) {
        randomOptions.totalWeight += opt.weight
        // 将选项按照权重多次添加到选项集合中
        for (i <- 1 to opt.weight) {
          randomOptions.optsBuffer += opt.value
        }
      }
      randomOptions
    }
  }

  // 随机选项集合类
  class RandomOptions[T](opts: RanOpt[T]*) {
    var totalWeight = 0
    var optsBuffer = new ListBuffer[T]

    // 从选项集合中随机获取一个选项
    def getRandomOpt: T = {
      val randomNum: Int = new Random().nextInt(totalWeight)
      optsBuffer(randomNum)
    }
  }
}
